package com.xzj.common.plugins.disruptor.mq;

import com.alibaba.fastjson2.JSON;
import com.lmax.disruptor.RingBuffer;
import com.xzj.common.plugins.disruptor.domain.DisruptorMessage;
import lombok.extern.slf4j.Slf4j;

/**
 * @program: xzj-common
 * @ClassName: DisruptorMqProducer
 * @description: 消息生产者
 * @author: XiaZunJun
 * @create: 2024-11-19 10:44
 */
@Slf4j
public class DisruptorMqProducer<T> {

    private final RingBuffer<DisruptorMessage<T>> disruptorMq;

    public DisruptorMqProducer(RingBuffer<DisruptorMessage<T>> disruptorMq) {
        this.disruptorMq = disruptorMq;
    }

    public void send(T t){
        log.debug("record the message: {}", JSON.toJSONString(t));
        //获取下一个Event槽的下标
        long sequence = disruptorMq.next();
        try {
            //给Event填充数据
            DisruptorMessage<T> event = disruptorMq.get(sequence);
            event.setT(t);
            log.debug("往消息队列中添加消息：{}", event);
        } catch (Exception e) {
            log.error("往消息队列添加消息失败！",e);
        } finally {
            //发布Event，激活观察者去消费，将sequence传递给改消费者
            //注意最后的publish方法必须放在finally中以确保必须得到调用；如果某个请求的sequence未被提交将会堵塞后续的发布操作或者其他的producer
            disruptorMq.publish(sequence);
        }
    }
}
